package com.lz.dsp.task.consumer;

import com.google.gson.Gson;
import com.lz.ad.common.mq.kafka.KafkaAdGroupCons;
import com.lz.ad.common.mq.kafka.KafkaAdTopicCons;
import com.lz.ad.common.util.AppUtils;
import com.lz.common.model.canal.RowChangeInfo;
import com.lz.dsp.task.common.Config;
import com.lz.dsp.task.entity.Creative;
import com.lz.dsp.task.enums.EventTypeEnum;
import com.lz.dsp.task.service.OutdoorsStallService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.exception.ExceptionUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.HashMap;
import java.util.Map;
import java.util.UUID;

/**
 * 广告创意异步处理线程
 *
 * @author zhy
 * @date 2019年7月13日
 */
@Component
@Slf4j
public class DspCreativeConsumer {
    private static final Gson GSON = new Gson();

    @Autowired
    private OutdoorsStallService outdoorsStallService;

    @KafkaListener(id = KafkaAdGroupCons.SYNC_DSP_CREATIVE,
            topics = KafkaAdTopicCons.CANAL_DSP_CREATIVE,
            containerFactory = Config.KAFKA_LISTENER_CONTAINER_FACTORY_SYNC_DSP_CREATIVE)
    public void processDspCreativeMessage(ConsumerRecord<String, String> cr, Acknowledgment ack) {
        String uuid = UUID.randomUUID().toString();
        log.info("uuid={}, receiver: key:{}, value:{} | partition:{} | offset:{} | thread:{}", uuid, cr.key(), cr.value(), cr.partition(), cr.offset(), Thread.currentThread().getName());
        String messageStr = cr.value();
        boolean handleResult = false;
        try {
            log.info("uuid={}, rowChangeInfo:{}", uuid, messageStr);
            // 转换成 行变更信息 对象
            RowChangeInfo rowChangeInfo = GSON.fromJson(messageStr, RowChangeInfo.class);

            // 判断rowChangeInfo是否为空
            if (rowChangeInfo == null) {
                log.info("uuid={}, rowChangeInfo is null", uuid);
                return; // 不需要重新入队
            }
            EventTypeEnum eventTypeEnum = EventTypeEnum.getEnum(rowChangeInfo.getEventType());
            if (eventTypeEnum == null) {
                log.info("uuid={}, eventTypeEnum is null", uuid);
                return;
            }
            log.info("uuid={}, eventTypeEnum:{}", uuid, eventTypeEnum.value());

            //新增创意-初始化档位表、标记附件表文件使用状态为已使用
            //删除创意-删除档位数据、标记附件表文件使用状态为未使用
            Creative entity = null;
            // delete 需要获取变更前的数据
            if (EventTypeEnum.DELETE.equals(eventTypeEnum)) {
                // 获取需要处理的对象信息
                entity = GSON.fromJson(rowChangeInfo.getBefore(), Creative.class);
            } else {
                // 获取需要处理的对象信息
                entity = GSON.fromJson(rowChangeInfo.getAfter(), Creative.class);
            }

            // 对象为空
            if (entity == null || AppUtils.isEmpty(entity.getId())) {
                return;
            }
            log.info("uuid={}, entity:{}", uuid, entity);

            handleResult = handle(uuid, entity, eventTypeEnum, rowChangeInfo.getChangeColumn());
        } catch (Exception e) {
            handleResult = false;
            e.printStackTrace();
            log.error("uuid={}, Exception={}", uuid, ExceptionUtils.getFullStackTrace(e));
        } finally {
            if (handleResult) {
                log.info("uuid={}, 消息处理成功，手动ack", uuid);
                // 手动 ack
                ack.acknowledge();
            } else {
                log.info("uuid={}, 消息处理失败重新入队 requeue", uuid);
                // 消息重新入队
            }
        }
    }

    private boolean handle(String uuid, Creative entity, EventTypeEnum eventTypeEnum, String changeColumns) {
        // 变更字段
        Map<String, Object> changeColumnMap = new HashMap<>(16);
        if (AppUtils.isNotEmpty(changeColumns)) {
            String[] changeColArr = changeColumns.split(",");
            for (String col : changeColArr) {
                if (AppUtils.isNotEmpty(col)) {
                    changeColumnMap.put(col.trim(), "Y");
                }
            }
        }

        return outdoorsStallService.handle(uuid, entity, eventTypeEnum, changeColumnMap);
    }
}
